package com.gjy.pulsar.simple;

import com.gjy.pulsar.domain.NlMessage;
import com.gjy.pulsar.util.PulsarUtil;
import org.apache.pulsar.client.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-08-02 10:14:14
 */
public class SimpleConsumer {
    private static final Logger log = LoggerFactory.getLogger(SimpleConsumer.class);

    public static void main(String[] args) throws PulsarClientException {
        PulsarClient client = PulsarUtil.getClient();
        Consumer<NlMessage> consumer = client.newConsumer(Schema.JSON(NlMessage.class))
                .topic("persistent://public/default/BG_GROUP_MIXMARKET_IN_11")
                .subscriptionName("my_subscription")
                .subscribe();
        while (true) {
            Message<NlMessage> message = consumer.receive();
            try {
                log.info("msg: {}", new String(message.getData(), StandardCharsets.UTF_8));
                consumer.acknowledge(message);
            } catch (Exception e) {
                consumer.negativeAcknowledge(message);
                e.printStackTrace();
            }
        }
    }
}
